package com.atguigu.flink.chapter05.transform;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SplitDemo3 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
        /*
        主流：奇数
        侧输出流： 偶数
         */
        OutputTag<Integer> evenTag = new OutputTag<Integer>("even") {};
        OutputTag<Integer> greaterThreeTag = new OutputTag<Integer>("greater-three") {};
        SingleOutputStreamOperator<Integer> oddStream = source.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value,
                                       Context ctx,
                                       Collector<Integer> out) throws Exception {
                if (value % 2 == 1) {
                    out.collect(value);
                } else {
                    //放入侧输出流
                    //侧输出流输出数据
                    ctx.output(evenTag, value);
                }


                if (value > 3){
                    ctx.output(greaterThreeTag,value);
                }
            }
        });
        SideOutputDataStream<Integer> evenStream = oddStream.getSideOutput(evenTag);

        oddStream.print("odd");
        evenStream.print("even");

        SideOutputDataStream<Integer> greaterThreeStream = oddStream.getSideOutput(greaterThreeTag);
        greaterThreeStream.print("greaterThree");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


//public class SplitDemo3 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
//        /*
//        主流：奇数
//        侧输出流： 偶数
//         */
//
//        OutputTag<Integer> evenTag = new OutputTag<Integer>("even") {};
//        OutputTag<Integer> greaterThreeTag = new OutputTag<Integer>("greater-three") {};
//        SingleOutputStreamOperator<Integer> oddStream = source.process(new ProcessFunction<Integer, Integer>() {
//            @Override
//            public void processElement(Integer value,
//                                       Context ctx,
//                                       Collector<Integer> out) throws Exception {
//                if (value % 2 == 1) {
//                    out.collect(value);
//                } else {
//                    //放侧输出流
//                    //侧输出流输出数
//                    ctx.output(evenTag, value);
//                }
//
//                if (value > 3){
//                    ctx.output(greaterThreeTag,value);
//                }
//            }
//        });
//        SideOutputDataStream<Integer> evenStream = oddStream.getSideOutput(evenTag);
//
//        oddStream.print("odd");
//        evenStream.print("even");
//
//        SideOutputDataStream<Integer> greaterThreeStream = oddStream.getSideOutput(greaterThreeTag);
//
//        greaterThreeStream.print("greaterThree");
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class SplitDemo3 {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        DataStreamSource<Integer> source = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8);
//        /*
//        主流： 奇数
//        侧输出流： 偶数
//         */
//        OutputTag<Integer> evenTag = new OutputTag<Integer>("even") {};
//        OutputTag<Integer> greaterThreeTag = new OutputTag<Integer>("greater-three") {};
//        SingleOutputStreamOperator<Integer> oddStream = source.process(new ProcessFunction<Integer, Integer>() {
//            /*
//            流中每来一个元素，该方法执行一次
//            参数1 ： 需要处理的元素
//            参数2 ： 上下文对象 环境
//            返回值： 收集器
//             */
//            @Override
//            public void processElement(Integer value,
//                                       Context ctx,
//                                       Collector<Integer> out) throws Exception {
//                if (value % 2 == 1) {
//                    out.collect(value);
//                } else {
//                    //放入侧输出流
//                    //侧输出流输出数据
//                    ctx.output(evenTag, value);
//                }
//
//                if (value > 3){
//                    ctx.output(greaterThreeTag,value);
//                }
//            }
//        });
//
//        SideOutputDataStream<Integer> evenStream = oddStream.getSideOutput(evenTag);
//
//        oddStream.print("odd");
//        evenStream.print("even");
//
//        SideOutputDataStream<Integer> greaterThreeStream = oddStream.getSideOutput(greaterThreeTag);
//
//        greaterThreeStream.print("greaterThree");
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}